package com.smsc.headend.connector.netty.handler;


import cn.hutool.core.date.SystemClock;
import cn.hutool.core.lang.Tuple;
import cn.hutool.core.util.HexUtil;
import com.smsc.headend.common.thread.UDISThreadFactory;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.connector.bo.ConnectTaskResultDto;
import com.smsc.headend.connector.bo.RetryBo;
import com.smsc.headend.connector.bo.TaskContext;
import com.smsc.headend.connector.builder.LoggerBuilder;
import com.smsc.headend.connector.enums.states.ChannelState;
import com.smsc.headend.connector.manager.AssociationManager;
import com.smsc.headend.connector.manager.TCPClientManager;
import com.smsc.headend.connector.netty.NettyConfigHelper;
import com.smsc.headend.connector.service.DeviceConnectService;
import com.smsc.headend.connector.service.DeviceStatusService;
import com.smsc.headend.connector.service.LogTerminalAccessService;
import com.smsc.headend.connector.service.TerminalChannelService;
import com.smsc.headend.connector.service.protocol.DLMSPacketCodecService;
import com.smsc.headend.connector.service.protocol.impl.DLMSPacketCodecServiceImpl;
import com.smsc.headend.connector.utils.LogFileNameUtil;
import com.smsc.headend.connector.utils.NetUtil;
import com.smsc.headend.module.asset.entity.ComEntity;
import com.smsc.headend.module.protocol.cosem.util.ByteUtils;
import com.smsc.headend.module.task.dto.ConnectorTask;
import com.smsc.headend.module.task.dto.ConnectorTaskResult;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.concurrent.DefaultEventExecutor;
import io.netty.util.concurrent.EventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateUtils;
import org.slf4j.Logger;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;

import java.util.Optional;
import java.util.concurrent.*;

import static com.smsc.headend.common.enums.errorcode.TaskExecuteErrorCode.TASK_WAITING_CON_RESPONSE_OVERTIME;

/**
 * server handler
 */
@Slf4j
@ChannelHandler.Sharable
public class DLMSPacketHandler extends SimpleChannelInboundHandler {

    @Autowired
    DeviceConnectService deviceConnectService;
    @Autowired
    LogTerminalAccessService logTerminalAccessService;

    @Autowired
    @Qualifier("connectorReceiveTaskPool")
    ThreadPoolExecutor threadPoolExecutor;

    @Autowired
    @Qualifier("retryService")
    ScheduledExecutorService scheduledExecutorService;

    @Autowired
    AssociationManager associationManager;

    @Autowired
    TerminalChannelService terminalChannelService;

    @Autowired
    DLMSPacketCodecService dlmsPacketCodecService;

    @Autowired
    DeviceStatusService deviceStatusService;

    @Autowired
    RedisUtils redisUtils;

    @Value("${packet.resend.enable: true}")
    Boolean enableResend;

    @Value("${packet.resend.interval:45}")
    Integer resendInterval;

    @Value("${packet.resend.timeout:120}")
    Integer packetTimeout;

    @Autowired
    LogFileNameUtil logFileNameUtil;

    @Autowired
    TCPClientManager tcpClientManager;

    @Autowired
    MeterRegistry registry;
    @Autowired
    @Qualifier("dlmsTimer")
    Timer acceptTimer;


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        String netAddr = NetUtil.getIpFromChannel(ctx.channel());
        MDC.put("channel", netAddr);
        ComEntity comEntity = ctx.channel().attr(NettyConfigHelper.comEntityAttrKey).get();
        if (comEntity != null && comEntity.getComId() != -1L) {
            terminalChannelService.addReceiveRecord(comEntity);
        }
        ByteBuf buf = (ByteBuf) msg;
        byte[] packetByteArr = new byte[buf.readableBytes()];
        buf.readBytes(packetByteArr);
        LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(comEntity), this.getClass()).info("<--: {}", ByteUtils.byteToHexString(packetByteArr));
        try {
            ChannelState channelState = ctx.channel().attr(NettyConfigHelper.channelState).get();
            if (dlmsPacketCodecService.isHeartBeat(packetByteArr)) {
                byte[] responseBytes = dlmsPacketCodecService.getResponseBytes(packetByteArr);
                if (responseBytes != null) {
                    deviceConnectService.sendPacket2Channel(ctx.channel(), responseBytes);
                }
                if (ChannelState.Init.equals(channelState)) {
                    String hexResult = HexUtil.decodeHexStr(ByteUtils.byteToHexString(packetByteArr).replace(" ", ""));
                    if (hexResult.startsWith(DLMSPacketCodecServiceImpl.DRLI)) {
                        ctx.channel().attr(NettyConfigHelper.deviceLgcPkg).set(packetByteArr);
                        ctx.channel().attr(NettyConfigHelper.channelState).set(ChannelState.WaitingReg);
                    }
                    return;
                }
                if (ChannelState.Registered.equals(channelState)) {
                    threadPoolExecutor.submit(() -> {
                        dlmsPacketCodecService.heartBeatResp(ctx.channel(), comEntity);
                    });
                }
                return;
            } else {
                if (ChannelState.Init.equals(channelState)) {
                    ctx.channel().attr(NettyConfigHelper.deviceLgcPkg).set(packetByteArr);
                    ctx.channel().attr(NettyConfigHelper.channelState).set(ChannelState.WaitingReg);
                    Tuple logicalAddrFromLoggingFrame = dlmsPacketCodecService.getLogicalAddrFromLoggingFrame(packetByteArr);
                    if (logicalAddrFromLoggingFrame != null && logicalAddrFromLoggingFrame.get(1) != null) {
                        deviceConnectService.sendPacket2Channel(ctx.channel(), logicalAddrFromLoggingFrame.get(1));
                    }
                }
            }
            doBusiPacketAccept(ctx.channel(), packetByteArr);
        } catch (Exception e) {
            log.error("channelRead0 error", e);
        }
    }

    private void doBusiPacketAccept(Channel channel, byte[] result) {
        ChannelState channelState = channel.attr(NettyConfigHelper.channelState).get();
        if (ChannelState.Registered.equals(channelState)) {
            threadPoolExecutor.submit(() -> {
                Timer.Sample sample = Timer.start(registry);
                deviceConnectService.dlmsPacketAccept(channel, result);
                sample.stop(acceptTimer);
            });
        }
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        ctx.channel().attr(NettyConfigHelper.channelState).setIfAbsent(ChannelState.Init);
        super.channelRegistered(ctx);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            switch (event.state()) {
                case READER_IDLE:
                    break;
                case WRITER_IDLE:
                    RetryBo retryBo = ctx.channel().attr(NettyConfigHelper.resendPkg).get();
                    ComEntity comEntity = ctx.channel().attr(NettyConfigHelper.comEntityAttrKey).get();
                    ConnectorTask connectorTask = ctx.channel().attr(NettyConfigHelper.connectTaskRegular).get();
                    if (connectorTask == null) {
                        break;
                    }
                    if (ctx.channel().attr(NettyConfigHelper.retryFuture).get() != null) {
                        LoggerBuilder.getLogger(comEntity.getModuleSn(), this.getClass()).debug("write Idle,retry scheduled.");
                        break;
                    }
                    //如果重试未开启，直接设置超时schedule
                    if (!enableResend) {
                        ScheduledFuture<?> schedule = scheduledExecutorService.schedule(() -> {
                            try {
                                taskOverTime(ctx, comEntity, retryBo);
                            } catch (Exception e) {
                                log.error("timeOverException, {}", Optional.ofNullable(comEntity).orElse(new ComEntity()).getComId(), e);
                            }
                        }, packetTimeout - 15, TimeUnit.SECONDS);
                        ctx.channel().attr(NettyConfigHelper.retryFuture).set(schedule);
                        break;
                    }
                    if (retryBo == null) {
                        break;
                    }
                    LoggerBuilder.getLogger(comEntity.getModuleSn(), this.getClass()).info("Set retry delay 1st will send retry message soon");
                    ScheduledFuture<?> schedule = scheduledExecutorService.schedule(() -> {
                        resendMessage(ctx, comEntity, retryBo);
                    }, resendInterval * DateUtils.MILLIS_PER_SECOND - (SystemClock.now() - retryBo.getTv()), TimeUnit.MILLISECONDS);
                    ctx.channel().attr(NettyConfigHelper.retryFuture).set(schedule);
                    break;
                case ALL_IDLE:
                    break;
            }
        }
        super.userEventTriggered(ctx, evt);
    }

    private void taskOverTime(ChannelHandlerContext ctx, ComEntity comEntity, RetryBo retryBo) {
        ConnectorTask connectorTask = ctx.channel().attr(NettyConfigHelper.connectTaskRegular).get();
        ctx.channel().attr(NettyConfigHelper.resendPkg).set(null);
        ctx.channel().attr(NettyConfigHelper.retryFuture).set(null);
        if (connectorTask == null) {
            return;
        }
        String atomicTaskNo = retryBo == null ? connectorTask.getAtomicTaskNo() : retryBo.getAtomicTaskNo();
        deviceConnectService.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, atomicTaskNo, null, TASK_WAITING_CON_RESPONSE_OVERTIME.toString(), comEntity.getComId(), connectorTask.getFlag(), connectorTask.getTicketNumber()), comEntity);
        //设置connectorTaskRegular为null防止重复结束
        ctx.channel().attr(NettyConfigHelper.connectTaskRegular).set(null);
        String ipFromChannel = NetUtil.getIpFromChannel(ctx.channel());
        LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(comEntity), this.getClass()).info("taskOverTime");
        if (retryBo != null) {
            TaskContext taskContext = terminalChannelService.getTaskContext(ipFromChannel, retryBo.getDeviceAddress(), comEntity.getComId());
            //仅设置AA状态
            LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(comEntity), this.getClass()).info("remove association");
            associationManager.setAssociationInvalid(comEntity, ipFromChannel, retryBo.getDeviceAddress(), false);
            if (taskContext != null) {
                taskContext.refresh(retryBo.getAtomicTaskNo());
            }
        }
    }

    private void resendMessage(ChannelHandlerContext ctx, ComEntity comEntity, RetryBo bo) {
        Channel channel = ctx.channel();
        String loggerName = comEntity.getModuleSn();
        LoggerBuilder.getLogger(StringUtils.isEmpty(loggerName) ? "unknown-device" : loggerName, this.getClass()).info("time: {}, resend task to {},", bo.getRetryTime(), NetUtil.getIpFromChannel(channel));
        channel.writeAndFlush(org.bouncycastle.pqc.math.linearalgebra.ByteUtils.clone(bo.getPacket()));
        bo.setTv(SystemClock.now());
        bo.setRetryTime((short) (bo.getRetryTime() + 1));
        ctx.channel().attr(NettyConfigHelper.resendPkg).set(bo);
        ScheduledFuture<?> schedule = null;
        Integer passedSecond = bo.getRetryTime() * resendInterval;
        if (passedSecond + resendInterval >= packetTimeout) {
            int leftSeconds = packetTimeout - passedSecond;
            LoggerBuilder.getLogger(StringUtils.isEmpty(loggerName) ? "unknown-device" : loggerName, this.getClass()).info("will timeout soon in {}s", leftSeconds);
            schedule = scheduledExecutorService.schedule(() -> {
                taskOverTime(ctx, comEntity, bo);
            }, leftSeconds, TimeUnit.SECONDS);

        } else {
            schedule = scheduledExecutorService.schedule(() -> {
                LoggerBuilder.getLogger(StringUtils.isEmpty(loggerName) ? "unknown-device" : loggerName, this.getClass())
                        .info("ready to resend {} in {}s", bo.getRetryTime(), resendInterval);
                resendMessage(ctx, comEntity, bo);
            }, resendInterval, TimeUnit.SECONDS);
        }
        ctx.channel().attr(NettyConfigHelper.retryFuture).set(schedule);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        NettyConfigHelper.group.add(ctx.channel());
        log.debug("channelActive ip {}", NetUtil.getIpFromChannel(ctx.channel()));
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ComEntity com = ctx.channel().attr(NettyConfigHelper.comEntityAttrKey).get();
        LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(com), this.getClass()).info("DLMSChannelInactive:comId:{},addr:{}", com != null ? com.getComId() : "annoy-device", NetUtil.getIpFromChannel(ctx.channel()));
        logTerminalAccessService.saveBytesCount(ctx.channel(), 0, Boolean.FALSE);
        if (!NettyConfigHelper.isChannelRemovedBySystem(ctx.channel())) {
            threadPoolExecutor.execute(() -> {
                try {
                    deviceConnectService.deviceDisconnect(ctx.channel());
                } catch (Exception e) {
                    log.error("updateComStateFailed. comId:{}", Optional.ofNullable(com).orElse(new ComEntity()).getComId(), e);
                }
            });
        }
        NettyConfigHelper.group.remove(ctx.channel());
        deviceConnectService.unlockComEntity(ctx.channel());
        tcpClientManager.channelInactive(ctx.channel());
    }
}
